package com.milo.coding.stream;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;
import java.util.stream.Stream;

/**
 * @Auther milo
 * @Description
 * @Date 2021-01-17 15:40
 */
public class StreamTest {



    public static void main(String[] args) {
//        createStream();

//        mapTest();
//        flatMapTest();
//        filterTest();
//        distinctTest();

//        limitOrSkipTest();
//        parallelTest();
//        reduceTest();
//        mathTest();
//        groupByTest();
//        partitioningByTest();

//        generateTest();
//        findFirstTest();

        // 并行迭代器
        iteratorTest();
    }

    static AtomicInteger count = new AtomicInteger(0);

    private static void iteratorTest() {
        List<Integer> strList = new ArrayList<>();
        Spliterator spliterator = strList.spliterator();

        // 初始化数据
        for (int i = 0; i < 100; i++) {
            strList.add(i);
        }


        // tryAdvance 相当于普通迭代器iterator  串行处理
//        AtomicInteger num = new AtomicInteger(0);
//        while(true){
//            boolean flag = spliterator.tryAdvance((i) ->{
//                num.addAndGet((int)i);
//                System.out.println(i);
//            });
//            if(!flag){
//                break;
//            }
//        }
//        System.out.println(num);


        // trySplit将list分段，每段单独处理，为并行提供可能
//        AtomicInteger num = new AtomicInteger(0);
//        Spliterator s1 = spliterator.trySplit(); // spliterator:50-99   s1:0-49
//        Spliterator s2 = spliterator.trySplit(); // spliterator:75-99   s2:50-74
//        spliterator.forEachRemaining((i) ->{
//            num.addAndGet((int)i);
//            System.out.println("spliterator:"+i);
//        });
//        s1.forEachRemaining((i) ->{
//            num.addAndGet((int)i);
//            System.out.println("s1:"+i);
//        });
//        s2.forEachRemaining((i) ->{
//            num.addAndGet((int)i);
//            System.out.println("s2:"+i);
//        });
//        System.out.println("最终结果："+num);


        // 利用分段，开启多线程处理
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            run(spliterator.trySplit());
            return "future1 finished!";
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            run(spliterator.trySplit());
            return "future2 finished!";
        });
        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            run(spliterator);
            return "future3 finished!";
        });
        CompletableFuture<Void> combindFuture = CompletableFuture.allOf(future1, future2, future3);
        try {
            combindFuture.get(); //阻塞方法
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println("future1: " + future1.isDone() + " future2: " + future2.isDone() + " future3: " + future3.isDone());
        System.out.println("最终结果为：" + count);

    }

    public static void run(Spliterator s1) {
        final String threadName = Thread.currentThread().getName();
        System.out.println("线程" + threadName + "开始运行-----");
        s1.forEachRemaining(new Consumer() {
            @Override
            public void accept(Object o) {
                count.addAndGet((Integer)o);
            }
        });
        System.out.println("线程" + threadName + "运行结束-----");
    }

    // 找出第一个元素后就会停止遍历，相当于短路操作
    private static void findFirstTest() {
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        Optional<Integer> firstNumber = numberList.stream()
                .findFirst();
        System.out.println(firstNumber.orElse(-1));
    }

    // 生成自己的随机数流
    private static void generateTest() {
        Random random = new Random();
        Stream<Integer> generateRandom = Stream.generate(random::nextInt);
        generateRandom.limit(5).forEach(System.out::println);
        // 生成自己的 UUID 流
        Stream<UUID> generate = Stream.generate(UUID::randomUUID);
        generate.limit(5).forEach(System.out::println);
    }

    // partitioningBy：按条件分组
    private static void partitioningByTest() {
        List<Integer> ageList = Arrays.asList(11, 22, 13, 14, 25, 26);
        Map<Boolean, List<Integer>> ageMap = ageList.stream()
                .collect(Collectors.partitioningBy(age -> age > 18));
        System.out.println("未成年人：" + ageMap.get(false));
        System.out.println("成年人：" + ageMap.get(true));
    }

    // groupingBy：分组聚合，相当于mysql的group  by
    private static void groupByTest() {
        List<Integer> ageList = Arrays.asList(11, 22, 13, 14, 25, 26);
        Map<String, List<Integer>> ageGrouyByMap = ageList.stream()
                .collect(Collectors.groupingBy(age -> String.valueOf(age / 10)));
        ageGrouyByMap.forEach((k, v) -> {
            System.out.println("年龄" + k + "0多岁的有：" + v);
        });
    }

    //min
    //max
    //count
    //anyMatch
    //allMatch
    //noneMatch
    //findFirst
    //findAny
    //iterator
    //Statistics：统计
    private static void mathTest() {
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6);
        IntSummaryStatistics stats = list.stream().mapToInt(x -> x).summaryStatistics();
        System.out.println("最小值：" + stats.getMin());
        System.out.println("最大值：" + stats.getMax());
        System.out.println("个数：" + stats.getCount());
        System.out.println("和：" + stats.getSum());
        System.out.println("平均数：" + stats.getAverage());
    }

    // 累加器
    private static void reduceTest() {
        //reduce：累加器
        //reduce中返回的结果会作为下次累加器计算的第一个参数
        Optional accResult = Stream.of(1, 2, 3, 4).reduce((acc, item) -> {
            System.out.println("acc : " + acc);
            acc += item;
            System.out.println("item: " + item);
            System.out.println("acc+ : " + acc);
            System.out.println("--------");
            return acc;
        });
    }

    // 并行流
    private static void parallelTest() {
        Long resourse = LongStream.rangeClosed(0,1000000000L)
                .parallel().reduce(0,Long::sum);
        System.out.println(resourse);
    }

    // limit：获取前n个元素
    // skip：丢弃前n个元素
    private static void limitOrSkipTest() {
        List<Integer> ageList = Arrays.asList(11, 22, 13, 14, 25, 26);
        ageList.stream()
                .limit(3)
                .forEach(age -> System.out.print(age+","));//11，22,13
        System.out.println();
        ageList.stream()
                .skip(3)
                .forEach(age -> System.out.print(age+","));//14,25,26
    }

    // 去重
    private static void distinctTest() {
        List<String> list = Arrays.asList("AA", "BB", "CC", "BB", "CC", "AA", "AA");
        long l = list.stream().distinct().count();
        System.out.println("count:"+l);
        String output = list.stream().distinct().collect(Collectors.joining(","));
        System.out.println(output);
    }

    // 数据筛选
    private static void filterTest() {
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        List<Integer> collect = numberList.stream()
                .filter(number -> number % 2 == 0)
                .collect(Collectors.toList()); // 执行汇聚操作的时候才去进行filter操作
        collect.forEach(number -> System.out.print(number + ",")); //不能中断 return break
    }

    // 数据处理
    private static void flatMapTest() {
        Stream<List<Integer>> inputStream = Stream.of(
                Arrays.asList(1),
                Arrays.asList(2, 3),
                Arrays.asList(4, 5, 6)
        );
        List<Integer> collect = inputStream
                .flatMap((childList) -> childList.stream())
                .collect(Collectors.toList());
        collect.forEach(number -> System.out.print(number + ","));
    }

    // 数据处理
    private static void mapTest() {
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        // 映射成 2倍数字
        List<Integer> collect = numberList.stream()
                .map(number -> number * 2)
                .collect(Collectors.toList());
        collect.forEach(number -> System.out.print(number + ","));
        System.out.println();
        numberList.stream()
                .map(number -> "数字 " + number + ",")
                .forEach(number -> System.out.println(number));
    }

    // 创建数据源
    private static void createStream() {
        List<String> nameList = Arrays.asList("Darcy", "Chris", "Linda", "Sid", "Kim", "Jack", "Poul", "Peter");
        String[] nameArr = {"Darcy", "Chris", "Linda", "Sid", "Kim", "Jack", "Poul", "Peter"};
        // 集合获取 Stream 流
        Stream<String> nameListStream = nameList.stream();
        // 集合获取并行 Stream 流
        Stream<String> nameListStream2 = nameList.parallelStream();
        // 数组获取 Stream 流
        Stream<String> nameArrStream = Stream.of(nameArr);
        // 数组获取 Stream 流
        Stream<String> nameArrStream1 = Arrays.stream(nameArr);
        // 文件流获取 Stream 流
        BufferedReader bufferedReader = null;
        try {
            bufferedReader = new BufferedReader(new FileReader("README.md"));
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        Stream<String> linesStream = bufferedReader.lines();
        // 从静态方法获取流操作
        IntStream rangeStream = IntStream.range(1, 10);
        rangeStream.limit(10).forEach(num -> System.out.print(num+","));
        System.out.println();
        IntStream intStream = IntStream.of(1, 2, 3, 3, 4);
        intStream.forEach(num -> System.out.print(num+","));
    }
}
